package demoflink.watermark;

import demoflink.entity.WaterSensor;
import demoflink.function.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class WatermarkOutOfOrderedDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("node1:9092,node2:9092,node3:9092")
                .setGroupId("local")
                .setTopics("first")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
        env.setParallelism(2);

        // 指定水位线
        WatermarkStrategy<WaterSensor> strategy = WatermarkStrategy
                //乱序
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withIdleness(Duration.ofSeconds(4))
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor waterSensor, long l) {
                        System.out.println("watermark===="+waterSensor.getTs());
                        return waterSensor.getTs() * 1000;
                    }
                });
        OutputTag<WaterSensor> lateDataOutTag = new OutputTag<>("late-data", Types.POJO(WaterSensor.class));
        SingleOutputStreamOperator<String> sos = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource")
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(strategy)
                .keyBy(x -> x.getId())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(5))
                .sideOutputLateData(lateDataOutTag)
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<WaterSensor> iterable, Collector<String> collector) throws Exception {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        String windowStart = DateFormatUtils.format(start, "yyyy-MM-dd HH:mm:ss");
                        String windowEnd = DateFormatUtils.format(end, "yyyy-MM-dd HH:mm:ss");
                        long count = iterable.spliterator().estimateSize();
                        collector.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + "]包含 " + count + " 条数据=====>" + iterable.toString());

                    }
                });
        sos.getSideOutput(lateDataOutTag).print();
        sos.print();

        env.execute();
    }


}
